package com.wugui.datax.rpc.remoting.invoker.call;

import com.wugui.datax.rpc.remoting.net.params.XxlRpcFutureResponse;
import com.wugui.datax.rpc.remoting.net.params.XxlRpcResponse;
import com.wugui.datax.rpc.util.XxlRpcException;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;

/**
 * @author xuxueli 2018-10-22 18:31:42
 */
public class XxlRpcInvokeFuture implements Future<Object> {

  private static final ThreadLocal<XxlRpcInvokeFuture> THREAD_INVOKER_FUTURE = new ThreadLocal<>();
  private final XxlRpcFutureResponse futureResponse;

  public XxlRpcInvokeFuture(XxlRpcFutureResponse futureResponse) {
    this.futureResponse = futureResponse;
  }

  /**
   * get future
   *
   * @param type
   * @param <T>
   * @return
   */
  public static <T> Future<T> getFuture(Class<T> type) {
    Future<T> future = (Future<T>) THREAD_INVOKER_FUTURE.get();
    THREAD_INVOKER_FUTURE.remove();
    return future;
  }

  /**
   * set future
   *
   * @param future
   */
  public static void setFuture(XxlRpcInvokeFuture future) {
    THREAD_INVOKER_FUTURE.set(future);
  }

  /** remove future */
  public static void removeFuture() {
    THREAD_INVOKER_FUTURE.remove();
  }

  public void stop() {
    // remove-InvokerFuture
    futureResponse.removeInvokerFuture();
  }

  @Override
  public boolean cancel(boolean mayInterruptIfRunning) {
    return futureResponse.cancel(mayInterruptIfRunning);
  }

  // ---------------------- thread invoke future ----------------------

  @Override
  public boolean isCancelled() {
    return futureResponse.isCancelled();
  }

  @Override
  public boolean isDone() {
    return futureResponse.isDone();
  }

  @Override
  public Object get() throws ExecutionException, InterruptedException {
    try {
      return get(-1, TimeUnit.MILLISECONDS);
    } catch (TimeoutException e) {
      throw new XxlRpcException(e);
    }
  }

  @Override
  public Object get(long timeout, TimeUnit unit)
      throws InterruptedException, ExecutionException, TimeoutException {
    try {
      // future get
      XxlRpcResponse xxlRpcResponse = futureResponse.get(timeout, unit);
      if (xxlRpcResponse.getErrorMsg() != null) {
        throw new XxlRpcException(xxlRpcResponse.getErrorMsg());
      }
      return xxlRpcResponse.getResult();
    } finally {
      stop();
    }
  }
}
